#include "src/net/tcp_connection.h"

#include <errno.h>

#include <cassert>

#include "src/base/fio66_logger.h"
#include "src/net/channel.h"
#include "src/net/event_loop.h"
#include "src/net/socket.h"
#include "src/net/sockets_opt.h"

namespace fio66::net {
void DefaultConnectionCallback(const TcpConnectionPtr& conn) {
  LOGD << conn->LocalAddress().ToIpPort() << " -> "
       << conn->PeerAddress().ToIpPort() << " is "
       << (conn->Connected() ? "UP" : "DOWN");
  // do not call conn->forceClose(), because some users want to register message
  // callback only.
}

void DefaultMessageCallback(const TcpConnectionPtr&, Buffer* buf,
                                        Timestamp) {
  buf->RetrieveAll();
}

TcpConnection::TcpConnection(EventLoop* loop, const std::string& nameArg,
                             int sockfd, const InetAddress& localAddr,
                             const InetAddress& peerAddr)
    : loop_(CHECK_NOTNULL(loop)),
      name_(nameArg),
      state_(kConnecting),
      reading_(true),
      socket_(new Socket(sockfd)),
      channel_(new Channel(loop, sockfd)),
      local_addr_(localAddr),
      peer_addr_(peerAddr),
      high_water_mark_(64 * 1024 * 1024) {
  channel_->SetReadCallback(std::bind(&TcpConnection::HandleRead, this, _1));
  channel_->SetWriteCallback(std::bind(&TcpConnection::HandleWrite, this));
  channel_->SetCloseCallback(std::bind(&TcpConnection::HandleClose, this));
  channel_->SetErrorCallback(std::bind(&TcpConnection::HandleError, this));
  LOGD << "TcpConnection::ctor[" << name_ << "] at " << this
            << " fd=" << sockfd;
  socket_->SetKeepAlive(true);
}

TcpConnection::~TcpConnection() {
  LOGD << "TcpConnection::dtor[" << name_ << "] at " << this
            << " fd=" << channel_->Fd() << " state=" << StateToString();
  assert(state_ == kDisconnected);
}

bool TcpConnection::GetTcpInfo(struct tcp_info* tcpi) const {
  return socket_->GetTcpInfo(tcpi);
}

std::string TcpConnection::GetTcpInfoString() const {
  char buf[1024];
  buf[0] = '\0';
  socket_->GetTcpInfoString(buf, sizeof buf);
  return buf;
}

void TcpConnection::Send(const void* data, int len) {
  Send(std::string_view(static_cast<const char*>(data), len));
}

void TcpConnection::Send(std::string_view message) {
  if (state_ == kConnected) {
    if (loop_->IsInLoopThread()) {
      SendInLoop(message);
    } else {
      void (TcpConnection::*fp)(std::string_view message) =
          &TcpConnection::SendInLoop;
      loop_->RunInLoop(std::bind(fp,
                                 this,  // FIXME
                                 message));
      // std::forward<std::string>(message)));
    }
  }
}

// FIXME efficiency!!!
void TcpConnection::Send(Buffer* buf) {
  if (state_ == kConnected) {
    if (loop_->IsInLoopThread()) {
      SendInLoop(buf->Peek(), buf->ReadableBytes());
      buf->RetrieveAll();
    } else {
      void (TcpConnection::*fp)(std::string_view message) =
          &TcpConnection::SendInLoop;
      loop_->RunInLoop(std::bind(fp,
                                 this,  // FIXME
                                 buf->RetrieveAllAsString()));
      // std::forward<std::string>(message)));
    }
  }
}

void TcpConnection::SendInLoop(std::string_view message) {
  SendInLoop(message.data(), message.size());
}

void TcpConnection::SendInLoop(const void* data, size_t len) {
  loop_->AssertInLoopThread();
  ssize_t nwrote = 0;
  size_t remaining = len;
  bool faultError = false;
  if (state_ == kDisconnected) {
    LOGW << "disconnected, give up writing";
    return;
  }
  // if no thing in output queue, try writing directly
  if (!channel_->IsWriting() && output_buffer_.ReadableBytes() == 0) {
    nwrote = sockets::Write(channel_->Fd(), data, len);
    if (nwrote >= 0) {
      remaining = len - nwrote;
      if (remaining == 0 && write_complete_callback_) {
        loop_->QueueInLoop(
            std::bind(write_complete_callback_, shared_from_this()));
      }
    } else  // nwrote < 0
    {
      nwrote = 0;
      if (errno != EWOULDBLOCK) {
        LOGE << "TcpConnection::SendInLoop";
        if (errno == EPIPE || errno == ECONNRESET)  // FIXME: any others?
        {
          faultError = true;
        }
      }
    }
  }

  assert(remaining <= len);
  if (!faultError && remaining > 0) {
    size_t oldLen = output_buffer_.ReadableBytes();
    if (oldLen + remaining >= high_water_mark_ && oldLen < high_water_mark_ &&
        high_water_mark_callback_) {
      loop_->QueueInLoop(std::bind(high_water_mark_callback_,
                                   shared_from_this(), oldLen + remaining));
    }
    output_buffer_.Append(static_cast<const char*>(data) + nwrote, remaining);
    if (!channel_->IsWriting()) {
      channel_->EnableWriting();
    }
  }
}

void TcpConnection::Shutdown() {
  // FIXME: use compare and swap
  if (state_ == kConnected) {
    SetState(kDisconnecting);
    // FIXME: shared_from_this()?
    loop_->RunInLoop(std::bind(&TcpConnection::ShutdownInLoop, this));
  }
}

void TcpConnection::ShutdownInLoop() {
  loop_->AssertInLoopThread();
  if (!channel_->IsWriting()) {
    // we are not writing
    socket_->ShutDownWrite();
  }
}

// void TcpConnection::shutdownAndForceCloseAfter(double seconds)
// {
//   // FIXME: use compare and swap
//   if (state_ == kConnected)
//   {
//     SetState(kDisconnecting);
//     loop_->RunInLoop(std::bind(&TcpConnection::shutdownAndForceCloseInLoop,
//     this, seconds));
//   }
// }

// void TcpConnection::shutdownAndForceCloseInLoop(double seconds)
// {
//   loop_->AssertInLoopThread();
//   if (!channel_->IsWriting())
//   {
//     // we are not writing
//     socket_->ShutdownWrite();
//   }
//   loop_->runAfter(
//       seconds,
//       makeWeakCallback(shared_from_this(),
//                        &TcpConnection::ForceCloseInLoop));
// }

void TcpConnection::ForceClose() {
  // FIXME: use compare and swap
  if (state_ == kConnected || state_ == kDisconnecting) {
    SetState(kDisconnecting);
    loop_->QueueInLoop(
        std::bind(&TcpConnection::ForceCloseInLoop, shared_from_this()));
  }
}

void TcpConnection::ForceCloseWithDelay(double seconds) {
  if (state_ == kConnected || state_ == kDisconnecting) {
    SetState(kDisconnecting);
    std::weak_ptr weak = shared_from_this();
    loop_->RunAfter(seconds, [weak]() {
      if (auto sharead = weak.lock()) {
        sharead->ForceClose();
      }
    });  // not ForceCloseInLoop to
         // avoid race condition
  }
}

void TcpConnection::ForceCloseInLoop() {
  loop_->AssertInLoopThread();
  if (state_ == kConnected || state_ == kDisconnecting) {
    // as if we received 0 byte in HandleRead();
    HandleClose();
  }
}

const char* TcpConnection::StateToString() const {
  switch (state_) {
    case kDisconnected:
      return "kDisconnected";
    case kConnecting:
      return "kConnecting";
    case kConnected:
      return "kConnected";
    case kDisconnecting:
      return "kDisconnecting";
    default:
      return "unknown state";
  }
}

void TcpConnection::SetTcpNoDelay(bool on) { socket_->SetTcpNoDelay(on); }

void TcpConnection::StartRead() {
  loop_->RunInLoop(std::bind(&TcpConnection::StartReadInLoop, this));
}

void TcpConnection::StartReadInLoop() {
  loop_->AssertInLoopThread();
  if (!reading_ || !channel_->IsReading()) {
    channel_->EnableReading();
    reading_ = true;
  }
}

void TcpConnection::StopRead() {
  loop_->RunInLoop(std::bind(&TcpConnection::StopReadInLoop, this));
}

void TcpConnection::StopReadInLoop() {
  loop_->AssertInLoopThread();
  if (reading_ || channel_->IsReading()) {
    channel_->DisableReading();
    reading_ = false;
  }
}

void TcpConnection::ConnectEstablished() {
  loop_->AssertInLoopThread();
  assert(state_ == kConnecting);
  SetState(kConnected);
  channel_->Tie(shared_from_this());
  channel_->EnableReading();

  connection_callback_(shared_from_this());
}

void TcpConnection::ConnectDestroyed() {
  loop_->AssertInLoopThread();
  if (state_ == kConnected) {
    SetState(kDisconnected);
    channel_->DisableAll();

    connection_callback_(shared_from_this());
  }
  channel_->Remove();
}

void TcpConnection::HandleRead(Timestamp receiveTime) {
  loop_->AssertInLoopThread();
  int savedErrno = 0;
  ssize_t n = input_buffer_.ReadFd(channel_->Fd(), &savedErrno);
  if (n > 0) {
    message_callback_(shared_from_this(), &input_buffer_, receiveTime);
  } else if (n == 0) {
    HandleClose();
  } else {
    errno = savedErrno;
    LOGE << "TcpConnection::HandleRead";
    HandleError();
  }
}

void TcpConnection::HandleWrite() {
  loop_->AssertInLoopThread();
  if (channel_->IsWriting()) {
    ssize_t n = sockets::Write(channel_->Fd(), output_buffer_.Peek(),
                               output_buffer_.ReadableBytes());
    if (n > 0) {
      output_buffer_.Retrieve(n);
      if (output_buffer_.ReadableBytes() == 0) {
        channel_->DisableWriting();
        if (write_complete_callback_) {
          loop_->QueueInLoop(
              std::bind(write_complete_callback_, shared_from_this()));
        }
        if (state_ == kDisconnecting) {
          ShutdownInLoop();
        }
      }
    } else {
      LOGE << "TcpConnection::HandleWrite";
      // if (state_ == kDisconnecting)
      // {
      //   ShutdownInLoop();
      // }
    }
  } else {
    LOGD << "Connection fd = " << channel_->Fd() << " is down, no more writing";
  }
}

void TcpConnection::HandleClose() {
  loop_->AssertInLoopThread();
  LOGD << "fd = " << channel_->Fd() << " state = " << StateToString();
  assert(state_ == kConnected || state_ == kDisconnecting);
  // we don't close fd, leave it to dtor, so we can find leaks easily.
  SetState(kDisconnected);
  channel_->DisableAll();

  TcpConnectionPtr guardThis(shared_from_this());
  connection_callback_(guardThis);
  // must be the last line
  close_callback_(guardThis);
}

void TcpConnection::HandleError() {
  int err = sockets::GetSocketError(channel_->Fd());
  LOGE << "TcpConnection::HandleError [" << name_ << "] - SO_ERROR = " << err
       << " " << strerror_tl(err);
}

}  // namespace fio66::net
